package com.personal.gateway.config;

import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.model.*;
import com.personal.gateway.constant.FlinkCommandConstantOld;
import com.personal.gateway.dao.SavePointDao;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.factory.annotation.Value;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author xuanqisong
 * @date 2021/5/27 18:29
 * @document
 */
@Configuration
public class OssConfig {

    @Value("${oss.endPoint}")
    private String endPoint;

    @Value("${oss.accessKeyId}")
    private String accessKeyId;

    @Value("${oss.accessKeySecret}")
    private String accessKeySecret;

    @Value("${oss.bucketName}")
    private String bucketName;

    public String getEndPoint() {
        return endPoint;
    }

    public void setEndPoint(String endPoint) {
        this.endPoint = endPoint;
    }

    public String getAccessKeyId() {
        return accessKeyId;
    }

    public void setAccessKeyId(String accessKeyId) {
        this.accessKeyId = accessKeyId;
    }

    public String getAccessKeySecret() {
        return accessKeySecret;
    }

    public void setAccessKeySecret(String accessKeySecret) {
        this.accessKeySecret = accessKeySecret;
    }

    public String getBucketName() {
        return bucketName;
    }

    public void setBucketName(String bucketName) {
        this.bucketName = bucketName;
    }

    public List<SavePointDao> getSavePointList(String env, String jobName) {
        OSS oss = new OSSClientBuilder().build(this.endPoint, this.accessKeyId, this.accessKeySecret);
        List<SavePointDao> tempSavePointDaoDirList = new ArrayList<>();
        List<SavePointDao> tempSavePointDaoFileList = new ArrayList<>();
        // 查找所有savepoint文件
        final int maxKeys = 200;
        String nextContinuationToken = null;
        ListObjectsV2Result result;
        do {
            ListObjectsV2Request listObjectsV2Request = new ListObjectsV2Request(this.bucketName)
                    .withMaxKeys(maxKeys);
            listObjectsV2Request.setContinuationToken(nextContinuationToken);
            listObjectsV2Request.setPrefix(FlinkCommandConstantOld.SAVE_POINT_PREFIX + "/" + env + "/" + jobName + "/");

            result = oss.listObjectsV2(listObjectsV2Request);

            for (OSSObjectSummary objectSummary : result.getObjectSummaries()) {
                SavePointDao savePointDao = new SavePointDao();
                savePointDao.setFilePath(objectSummary.getKey());
                savePointDao.setJobName(jobName);
                savePointDao.setFileTime(objectSummary.getLastModified());
                if (objectSummary.getKey().endsWith("/")) {
                    tempSavePointDaoDirList.add(savePointDao);
                } else {
                    tempSavePointDaoFileList.add(savePointDao);
                }
            }

            nextContinuationToken = result.getNextContinuationToken();

        } while (result.isTruncated());
        return tempSavePointDaoDirList
                .stream()
                .peek(
                        savePointDao -> {
                            List<SavePointDao> tempList = tempSavePointDaoFileList.stream().filter(savePointDaoFile -> savePointDaoFile.getFilePath().startsWith(savePointDao.getFilePath())).collect(Collectors.toList());
                            if (tempList.size() > 0) {
                                savePointDao.setFileTime(tempList.get(0).getFileTime());
                                savePointDao.setOk(true);
                            }
                        }
                )
                .filter(SavePointDao::isOk)
                .sorted(Comparator.comparingLong(SavePointDao::getFileTimeStamp).reversed())
                .collect(Collectors.toList());
    }

}
